package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Stream_Stop {
  def main(args: Array[String]): Unit = {
    //TODO SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    ssc.checkpoint("cp")
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)
    //有状态数据操作
    //所谓的有状态，其实就是将采集周期的计算结果保存到检查点中。工作中都是保存在第三方的数据库中例如：Redis
    val wordDS: DStream[(String, Int)] = dstream.map((_, 1))

    wordDS.print()

    ssc.start()

    //TODO 什么场景中才需要停止数据操作。1、业务变了。2、技术升级。
    // SpatrkStreaming操作流程和业务的变化没有本质的联系，所以关闭不应该放在一起（线程）
    //因此一般判断关闭是新开一个线程，周期性读取中间件的方式来查看是否进行关闭，这个中间件就要根据实际情况进行指定了
    new Thread(

      new Runnable {
        override def run(): Unit = {
          //周期性读取【中间件】的关闭标记位
          while (true){

            val closeFlg = false
            while (closeFlg){
              Thread.sleep(6000)
              //第二个参数就是优雅的关闭
              ssc.stop(true,true)
              System.exit(0)
            }

          }

        }
      }
    ).start()

    ssc.awaitTermination()
  }

}
